package com.relytosoft.mizar.analyze.handle.service.impl;

import com.alibaba.fastjson.JSON;
import com.relytosoft.dubhe.common.elasticsearch.ElasticsearchUtil;
import com.relytosoft.dubhe.common.kafka.producer.KafkaProducer;
import com.relytosoft.dubhe.common.redis.utils.RedisUtil;
import com.relytosoft.mizar.analyze.base.constant.AyCacheConstants;
import com.relytosoft.mizar.analyze.base.entity.algo.AYFace;
import com.relytosoft.mizar.analyze.base.entity.algo.AYVideoSlice;
import com.relytosoft.mizar.analyze.base.entity.algo.AyControlFace;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.AyControlAlarmDto;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlDataReq;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlFaceControlReq;
import com.relytosoft.mizar.analyze.base.enums.IndexNameTypeEnum;
import com.relytosoft.mizar.analyze.common.utils.IndexUtil;
import com.relytosoft.mizar.analyze.handle.feign.AlarmFeignService;
import com.relytosoft.mizar.analyze.handle.service.ControlPlanService;
import com.relytosoft.mizar.analyze.handle.service.FaceListenerService;
import com.relytosoft.mizar.analyze.handle.service.UploadImageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

import static com.relytosoft.mizar.analyze.base.constant.MediaMqConstants.UNIFY_ALARM_TOPIC;
import static com.relytosoft.mizar.analyze.handle.constant.ErrorConstants.*;

/**
 * @author wss
 * @version 1.0
 * @date 2022/5/16 23:02
 */
@Service
public class FaceListenerServiceImpl implements FaceListenerService<BtdlDataReq, BtdlFaceControlReq> {

    protected Logger log = LoggerFactory.getLogger(this.getClass());

    static ThreadPoolExecutor threadPoolExecutor;

    static {
        threadPoolExecutor = new ThreadPoolExecutor(
                (int) (Runtime.getRuntime().availableProcessors() * 1.5),
                Runtime.getRuntime().availableProcessors() * 3,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<Runnable>(10000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

    @Autowired
    ControlPlanService controlPlanService;
    @Autowired
    private ElasticsearchUtil elasticsearchUtil;
    @Autowired
    private KafkaProducer kafkaProducer;
    @Autowired
    private UploadImageService uploadImageService;
    @Autowired
    private AlarmFeignService alarmFeignService;

    @Override
    public Boolean handleRawData(BtdlDataReq baseAlgoReq) {
        try {
            List<AYVideoSlice> videoSlices = baseAlgoReq.getVideoSliceListObject().getVideoSlice();
            if (videoSlices.isEmpty()) {
                return false;
            }
            videoSlices.stream().forEach(v -> {
                List<AYFace> faceObject = v.getFaceList().getFaceObject();
                rawDateUrlReplace(faceObject);
            });
            return true;
        } catch (Exception e) {
            log.error("{},--", FACE_HANDLE_RAW_DATA_ERROR, e);
            return false;
        }
    }

    /**
     * 人脸任务原始上报数据 图片url替换
     *
     * @param ayFaces
     */
    public void rawDateUrlReplace(List<AYFace> ayFaces) {
        ayFaces.stream().forEach(v -> {
            CompletableFuture<Void> imageReplace = CompletableFuture.runAsync(() -> v.setImage(uploadImageService.base64Upload(v.getImage())));
            CompletableFuture<Void> thumbReplace = CompletableFuture.runAsync(() -> v.setThumb(uploadImageService.base64Upload(v.getThumb())));
            CompletableFuture.allOf(imageReplace, thumbReplace).join();
        });
    }

    @Override
    public void handleIOData(BtdlDataReq baseAlgoReq) {
        try {
            List<AYVideoSlice> videoSlice = baseAlgoReq.getVideoSliceListObject().getVideoSlice();
            videoSlice.stream().forEach(v -> {
                List<AYFace> ayFaces = v.getFaceList().getFaceObject();
                threadPoolExecutor.execute(() -> saveFaceDataToDb(ayFaces));
                threadPoolExecutor.execute(() -> saveToRedis(ayFaces.size(), AyCacheConstants.FACE_TOTAL_STATISTICS));
            });
        } catch (Exception e) {
            log.error("{},--", FACE_HANDLE_IO_DATA_ERROR, e);
        }
    }

    /**
     * 保存数据到人脸es redis
     *
     * @param ayFaces
     */
    void saveFaceDataToDb(List<AYFace> ayFaces) {
        if (ayFaces.isEmpty()) {
            return;
        }
        String vehicleIndex = IndexUtil.toMonthIndex(IndexNameTypeEnum.AY_FACE.getMessage());
        try {
            elasticsearchUtil.addDocmentByBatch(vehicleIndex, null, ayFaces);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存人脸统计总数到redis
     *
     * @param num
     * @param statisticsName
     */
    void saveToRedis(int num, String statisticsName) {
        Integer newSum;
        Object oldSum = RedisUtil.get(statisticsName);
        newSum = Objects.isNull(oldSum)
                ? num
                : Integer.parseInt(oldSum.toString()) + num;
        RedisUtil.set(statisticsName, newSum);
    }

    @Override
    public void handleBusinessData(BtdlDataReq baseAlgoReq) {
        //
    }

    @Override
    public Boolean handleControlRawData(List<BtdlFaceControlReq> algoControlReqs) {
        try {
            algoControlReqs.stream().forEach(algoControlReq -> {
                AyControlFace gaFace = algoControlReq.getgAFace();
                controlDateUrlReplace(gaFace);
            });
            return true;
        } catch (Exception e) {
            log.error("{},--", FACE_HANDLE_CONTROL_RAW_DATA_ERROR, e);
            return false;
        }
    }

    /**
     * 布控上报原始数据 图片url替换
     *
     * @param gaFace
     */
    public void controlDateUrlReplace(AyControlFace gaFace) {
        CompletableFuture<Void> imageReplace = CompletableFuture.runAsync(() -> gaFace.setImage(uploadImageService.rawUrlUpload(gaFace.getImage())));
        CompletableFuture<Void> thumbReplace = CompletableFuture.runAsync(() -> gaFace.setThumb(uploadImageService.rawUrlUpload(gaFace.getThumb())));
        CompletableFuture.allOf(imageReplace, thumbReplace).join();
    }

    @Override
    public void handleControlIOData(List<BtdlFaceControlReq> faceControlReqs) {
        try {
            threadPoolExecutor.execute(() -> saveFaceControlDataToEs(faceControlReqs));
        } catch (Exception e) {
            log.error("{},--", FACE_HANDLE_CONTROL_IO_DATA_ERROR, e);
        }
    }

    /**
     * 布控人脸数据保存到es redis
     *
     * @param faceControlReqs
     */
    void saveFaceControlDataToEs(List<BtdlFaceControlReq> faceControlReqs) {
        if (faceControlReqs.isEmpty()) {
            return;
        }
        String vehicleIndex = IndexUtil.toMonthIndex(IndexNameTypeEnum.AY_CONTROL_FACE.getMessage());
        try {
            elasticsearchUtil.addDocmentByBatch(vehicleIndex, null, faceControlReqs);
            saveToRedis(faceControlReqs.size(), AyCacheConstants.CONTROL_FACE_TOTAL_STATISTICS);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void handleControlBusinessData(BtdlFaceControlReq algoControlReq) {
        if (controlPlanService.controlFaceAlarm(algoControlReq)) {
            try {
                AyControlAlarmDto ayControlAlarmDto = BtdlFaceControlReq.buildControlAlarm(algoControlReq);
                String jsonString = JSON.toJSONString(ayControlAlarmDto);
                CompletableFuture<Void> alarmSave = CompletableFuture.runAsync(() -> kafkaProducer.send(UNIFY_ALARM_TOPIC, jsonString, true), threadPoolExecutor);
                CompletableFuture<Void> alarmReport = CompletableFuture.runAsync(() -> alarmFeignService.alarmReport(jsonString), threadPoolExecutor);
                CompletableFuture.allOf(alarmSave, alarmReport).join();
            } catch (Exception e) {
                log.error("{},--", FACE_HANDLE_CONTROL_BUSINESS_DATA_ERROR, e);
            }
        }
    }

}
